package com.shujia.spark.streaming

import org.apache.spark.sql.streaming.OutputMode
import org.apache.spark.sql.{DataFrame, Dataset, Row, SaveMode, SparkSession}

object Demo06StructuredStreaming {
  def main(args: Array[String]): Unit = {

    val spark: SparkSession = SparkSession
      .builder()
      .appName(this.getClass.getSimpleName.replace("$", ""))
      .master("local[2]")
      .config("spark.sql.shuffle.partitions", "2")
      .getOrCreate()

    // 通过SparkSession readStream加载流式数据
    val textDF: DataFrame = spark
      .readStream
      .format("socket")
      .option("host", "master")
      .option("port", 8888)
      .load()

    textDF.createOrReplaceTempView("textDFTb")

    // 通过SQL或者是DSL完成实时数据处理
    // 状态会自动进行维护
    // 延时看起来比较低，实际上还是微批处理，只不过批次非常小是ms级别
    spark
      .sql(
        """
          |select t1.word
          |       ,count(*) as cnt
          |from (
          | select explode(split(value,",")) as word
          | from textDFTb
          |) t1 group by t1.word
          |""".stripMargin)
      //      .show() // 流处理不能直接show
      .writeStream
      .outputMode("complete")
      .format("console") // 将结果作为流进行输出
//      .start() // 启动任务
//      .awaitTermination() // 等待停止

    import spark.implicits._
    import org.apache.spark.sql.functions._
    // 亦可以通过DSL来处理数据
    textDF
      .select(explode(split($"value",",")) as "word")
      .groupBy($"word")
      .agg(count("*") as "cnt")
      .writeStream
      /**
      * OutputMode有哪些方式：
      * 1、Complete：每次都将所有结果进行输出，只能用于包含了聚合操作的流处理
      * 2、Append：每次将新增的rows进行输出，只能用于不包含聚合操作的流处理
      * 3、Update：所有场景都适用，每次只会将新增或者有变化的结果进行输出
        */
      .outputMode(OutputMode.Update())
      .format("console") // 将结果作为流进行输出
    //      .start() // 启动任务
    //      .awaitTermination() // 等待停止


    textDF
      .writeStream
      .outputMode(OutputMode.Append())
      .format("console")
//      .start()
//      .awaitTermination()

    // 将wordCnt的结果写入MySQL
    textDF
      .select(explode(split($"value", ",")) as "word")
      .groupBy($"word")
      .agg(count("*") as "cnt")
      .writeStream
      .outputMode(OutputMode.Complete())
      // 将每个批次转换成DataFrame进行处理
      .foreachBatch((df:Dataset[Row], l:Long)=>{
        println(l) // 批次的编号
        df.write
          .format("jdbc")
          .option("url", "jdbc:mysql://master:3306/student")
          .option("dbtable", "word_cnt")
          .option("user", "root")
          .option("password", "123456")
          .option("truncate", "true") // 以truncate的方式实现Overwrite，避免表结构发生变化
          .mode(SaveMode.Overwrite)
          .save()
      })
      .start()
      .awaitTermination()
  }

}
